#ifndef MQSERVER_MQ_BROKER_HPP_
#define MQSERVER_MQ_BROKER_HPP_
#include "mq_channel.hpp"
#include "muduo/net/InetAddress.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "../mqcomm/mq_logger.hpp"
#include "../mqcomm/proto/mq_msg.pb.h"
#include "../mqcomm/proto/mq_protocol.pb.h"

#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "mq_host.hpp"


namespace mqserver 
{
using std::placeholders::_1;
using std::placeholders::_2;
using std::placeholders::_3;

const std::string listenAddr = "0.0.0.0";
const std::string host_name = "host1";
const std::string dbfile = "/meta.db";

using GoogleMessagePtr = std::shared_ptr<google::protobuf::Message>;

class Server
{
public:
    Server(uint16_t port, const std::string basedir)
        : server_(&baseloop_, muduo::net::InetAddress(listenAddr, port), 
            "Server", muduo::net::TcpServer::Option::kReusePort)
        , dispatcher_(std::bind(&Server::onUnknownMessage, this, _1, _2, _3))
        , codec_(std::make_shared<ProtobufCodec>(
            std::bind(&ProtobufDispatcher::onProtobufMessage, &dispatcher_, _1, _2, _3)))
        , virtual_host_(std::make_shared<VirtualHost>(host_name, basedir, basedir + dbfile))
        , consumer_mg_(std::make_shared<ConsumerManager>())
        , connection_mg(std::make_shared<ConnectionManager>())
        , threadpool_(std::make_shared<pool::ThreadPool>())
    {
        // 启动线程池
        threadpool_->start();      

        // 针对历史所有的队列, 初始化队列的消费者结构,
        MsgQueuePtrMap queues_map = virtual_host_->getAllQueues();
        for (auto& [qname, queue] : queues_map) {
            consumer_mg_->initQueueConsumer(qname);
        }

        // 注册业务请求处理函数
        dispatcher_.registerMessageCallback<proto::protocol::openChannelRequest>(
            std::bind(&Server::onOpenChannel, this, _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::closeChannelRequest>(
            std::bind(&Server::onCloseChannel, this, _1, _2, _3)
        );

        dispatcher_.registerMessageCallback<proto::protocol::declareExchangeRequest>(
            std::bind(&Server::onDeclareExchange, this, _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::deleteExchangeRequest>(
            std::bind(&Server::onDeleteExchange, this, _1, _2, _3)
        );

        dispatcher_.registerMessageCallback<proto::protocol::declareQueueRequest>(
            std::bind(&Server::onDeclareQueue, this, _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::deleteQueueRequest>(
            std::bind(&Server::onDeleteQueue, this, _1, _2, _3)
        );

        dispatcher_.registerMessageCallback<proto::protocol::queueBindRequest>(
            std::bind(&Server::onQueueBind, this, _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::queueUnBindRequest>(
            std::bind(&Server::onQueueUnBind, this, _1, _2, _3)
        );    

        dispatcher_.registerMessageCallback<proto::protocol::basicPublishRequest>(
            std::bind(&Server::onBasicPublish, this,  _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::basicAckRequest>(
            std::bind(&Server::onBasicAck, this, _1, _2, _3)
        );         

        dispatcher_.registerMessageCallback<proto::protocol::basicConsumeRequest>(
            std::bind(&Server::onBasicConsume, this, _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::basicCancelRequest>(
            std::bind(&Server::onBasicCancel, this, _1, _2, _3)
        );            

        // 设置消息默认回调处理函数, 通过onMessage将dispatcher_注册的业务请求处理函数进行回调
        server_.setMessageCallback(std::bind(&ProtobufCodec::onMessage, codec_.get(), _1, _2, _3));
        // 默认连接回调处理函数
        server_.setConnectionCallback(std::bind(&Server::onConnection, this, _1));
    }
    void start()
    {
        server_.start();
        baseloop_.loop();
    }
private:
    // 打开/关闭信道
    void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, 
                       const openChannelRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onOpenChannel");
		if(self_conn.get()) self_conn->openChannel(message);
    }
    void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, 
                       const closeChannelRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onCloseChannel");
		if(self_conn.get()) self_conn->closeChannel(message);
    }
    // 声明/删除交换机
    void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, 
                           const declareExchangeRequestPtr& message, muduo::Timestamp)
    {
        // 1. 先通过 TcpConnectionPtr 查连接
		ConnectionPtr self_conn = getConnectionMapped(conn, "onDeclareExchange");
		if(self_conn.get() == nullptr) return;
        // 2. 再通过 self_conn 查信道
		ChannelPtr channel = getChannel(self_conn, message, "onDeclareExchange");
		if(channel.get()) channel->declareExchange(message);
    }
    void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn,
                          const deleteExchangeRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onDeleteExchange");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onDeclareExchange");
		if(channel.get()) channel->deleteExchange(message);
    }
    // 声明/删除消息队列
    void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn,
                        const declareQueueRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onDeclareQueue");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onDeclareQueue");
		if(channel.get()) channel->declareQueue(message);
    }
    void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, 
                       const deleteQueueRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onDeclareQueue");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onDeleteQueue");
		if(channel.get()) channel->deleteQueue(message);
    }
    // 队列绑定/解绑
    void onQueueBind(const muduo::net::TcpConnectionPtr& conn, 
                     const queueBindRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onQueueBind");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onQueueBind");
		if(channel.get()) channel->QueueBind(message);
    }
    void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn,
                       const queueUnBindRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onQueueUnBind");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onQueueUnBind");
		if(channel.get()) channel->QueueUnbind(message);
    }
    // 消息发布/确认
    void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, 
                        const basicPublishRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onBasicPublish");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onBasicPublish");
		if(channel.get()) channel->basicPublish(message);
    }
    void onBasicAck(const muduo::net::TcpConnectionPtr& conn, 
                    const basicAckRequestPtr& message, muduo::Timestamp)
    {
  		ConnectionPtr self_conn = getConnectionMapped(conn, "onBasicPublish");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onBasicPublish");
		if(channel.get()) channel->basicAck(message);      
    }
    // 队列订阅/取消订阅
    void onBasicConsume(const muduo::net::TcpConnectionPtr& conn,
                        const basicConsumeRequestPtr& message, muduo::Timestamp)
    {
		ConnectionPtr self_conn = getConnectionMapped(conn, "onBasicConsume");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onBasicConsume");
		if(channel.get()) channel->basicConsume(message);
    }
    void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, 
                       const basicCancelRequestPtr& message, muduo::Timestamp)
    {
   		ConnectionPtr self_conn = getConnectionMapped(conn, "onBasicCancel");
		if(self_conn.get() == nullptr) return;
		ChannelPtr channel = getChannel(self_conn, message, "onBasicCancel");
		if(channel.get()) channel->basicCancel(message);     
    }

    void onConnection(const muduo::net::TcpConnectionPtr& conn)
    {
        if(conn->connected() == true) {
            // LOG_DEBUG << "The new connection is successfully established!\n";
            connection_mg->newConnection(codec_, conn, virtual_host_, consumer_mg_, threadpool_);
        }
        else {
            // LOG_DEBUG << "The new connection is closed\n";
            connection_mg->deleteConnection(conn);
        }
    }
    void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                          const GoogleMessagePtr& message, muduo::Timestamp)
    {
        LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        conn->shutdown();
    }
private:
	ConnectionPtr getConnectionMapped(const muduo::net::TcpConnectionPtr& conn, const std::string& action)
	{
        ConnectionPtr self_conn = connection_mg->getConnection(conn);
        if(self_conn.get() == nullptr) {
            LOG_DEBUG << "Can't find the connection object mapped to the connection when"
                      << action << '\n';
            conn->shutdown();           // 关闭信道
            return ConnectionPtr();
        }
        return std::move(self_conn);
    }
	template <typename Request>
	ChannelPtr getChannel(const ConnectionPtr& self_conn,
		                  const Request& message,
		                //    const std::shared_ptr<Request>& message,
		                  const std::string& action)
	{
        ChannelPtr channel = self_conn->getChannel(message->channel_id());
        if(channel.get() == nullptr) {
            LOG_DEBUG << "can't find the channel when " << action << '\n';
            return ChannelPtr();
        }
        return std::move(channel);
    }
private:
    muduo::net::EventLoop baseloop_;    // IO监控
    muduo::net::TcpServer server_;      // TCP服务器
    ProtobufDispatcher dispatcher_;     // 请求分发器(将不同类型的请求分发到匹配的处理函数)
    ProtobufCodecPtr codec_;            // 协议处理器(负责应用层协议的解析和封装)
    VirtualHostPtr virtual_host_;
    ConsumerManagerPtr consumer_mg_;       
    ConnectionManagerPtr connection_mg;
    pool::ThreadPoolPtr threadpool_;    
};

}


#endif